#coding:utf-8
from kafka import KafkaProducer
from kafka import KafkaConsumer
import threading
from pymongo import MongoClient, ReturnDocument
from urllib.parse import quote_plus
import json
import time
import string
import base64
import random
from random import choice
from azure.storage.blob import BlockBlobService, PublicAccess

class KafkaOperate(object):

    class TestMsg:
        def testMsg_to_str(testMsg):
            deviceId = "12345" #''.join(random.choices(string.ascii_letters + string.digits, k=8))
            dataTypes = "devicePic" #["deviceLog", "deviceData", "devicePic", "ClientInfo"]
            data = None
            dataType = dataTypes #choice(dataTypes)
            print(type(dataType))
            if dataType == "deviceLog":
                data = {"source":"client", "message":"something wrong.", "level":"error", "uploadTime":"2019/3/7 11:50:11"}
            elif dataType == "deviceData":
                data = {"data" : ["数据1", "数据2", "数据3"]}
            elif dataType == "devicePic":
                with open("C:/Users/Administrator/Pictures/500x250.png","rb") as f:
                    base64_data = base64.b64encode(f.read())
                data = {"data" : [base64_data]}
            elif dataType == "ClientInfo":
                data = {"mac":"d0:17:c2:01:e5:37", "ip":"192.168.8.72", "memory":"4G", "cpu":"2"}
            else :
            	data = {}
            return {'dataType': dataType, 'deviceId': deviceId, 'data': data}

    def __init__(self):
        uri = "mongodb://%s:%s@%s" % (quote_plus("robot"), quote_plus("robot#248#cloud"), "192.168.58.50:27010/ezrobot_cloud")
        self.client = MongoClient(uri)
        self.mongoDB = self.client['ezrobot_cloud']
        # self.collection = self.mongoDB['ezrobot_collection']

    # Creat Collection to mongodb
    def insertToCollection(self, document):
        result = self.collection.find_one({"_id": 'info_id'})
        if result is None:
           timestamp = time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))
           result = self.collection.insert_one({"_id" : "info_id", "sequence_value" : 0, "timestamp" : timestamp})
        document["sequence_value"] = int(self.getNextId())
        document["timestamp"] = time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))
        result = self.collection.insert_one(document)
        print("insert.inserted_id result is :", result.inserted_id)

    # send message
    def send(self):
        producer = KafkaProducer(bootstrap_servers=['192.168.58.57:9092'])
        testMsg = self.TestMsg()
        dataType = "devicePic"
        deviceId = "12345"
        with open("C:/Users/Administrator/Pictures/500x250.png","rb") as f:
            base64_data = base64.b64encode(f.read())
        data = {"data" : [base64_data.decode()]}
        testMsg = {'dataType': dataType, 'deviceId': deviceId, 'data': data}
        jsonStr = json.dumps(testMsg)
        msgByte = bytes(jsonStr, encoding = "utf8")
        future = producer.send("test-topic2", msgByte)
        result = future.get(timeout= 60)
        #print(result)
        timer = threading.Timer(5, fun_timer)
        timer.start()
        
    def writeToFile(self, type, json):
        if type == "pic":
            with open('F:/pic.txt', 'a') as p:
                p.writelines([json,'\n'])
        else:
            with open('F:/data.txt', 'a') as f:
                f.writelines([json,'\n'])

    # Classification processing data
    def processData(self, dataJson, realIp):
        try:
            jsonList = json.loads(dataJson, strict = False)
                        
            for jsonItem in jsonList:
                if "dataType" in jsonItem.keys() and jsonItem["dataType"] == "encoderData":
                    # Creat collection and insert record to mongoDB
                    self.collection = self.mongoDB["encoderData"]
                    jsonItem["realIp"] = realIp
                    self.insertToCollection(jsonItem)

                elif "dataType" in jsonItem.keys() and jsonItem["dataType"] == "productionData":
                    self.collection = self.mongoDB["productionData"]
                    jsonItem["realIp"] = realIp
                    self.insertToCollection(jsonItem)

                elif "dataType" in jsonItem.keys() and jsonItem["dataType"] == "robotData":
                    self.collection = self.mongoDB["robotData"]
                    jsonItem["realIp"] = realIp
                    self.insertToCollection(jsonItem)

                elif "dataType" in jsonItem.keys() and jsonItem["dataType"] == "clawData":
                    self.collection = self.mongoDB["clawData"]
                    jsonItem["realIp"] = realIp
                    self.insertToCollection(jsonItem)

                elif "dataType" in jsonItem.keys() and jsonItem["dataType"] == "pcData":
                    self.collection = self.mongoDB["pcData"]
                    jsonItem["realIp"] = realIp
                    self.insertToCollection(jsonItem)

                elif "dataType" in jsonItem.keys() and jsonItem["dataType"] == "pic":
                    # tempJsonItem = jsonItem
                    pass

                else :
                        print("Unexpected error! json = " , dataJson)
        except ValueError as e:
            #print("Decoding data has failed! json is " , jsonStr, e)
            pass

    # receive message and insert document to mongodb
    def receive(self):
        #consumer = KafkaConsumer('test-topic', group_id='test-transfer', bootstrap_servers= ['192.168.58.57:9092'])
        consumer = KafkaConsumer('test-transfer', group_id='test-transfer', bootstrap_servers= ['192.168.58.57:9092'])
        for msg in consumer:
            #jsonStr = str(msg.value, encoding = "utf-8")
            jsonStr = msg.value.decode("utf-8")
            #print("msg.offset == " , msg.offset, "  topic = ", msg.topic)
            #continue
            print("msg = " , jsonStr)
            
            if jsonStr == "":
                continue
            try:
                jsonObj = json.loads(jsonStr, strict = False)
                realIp = jsonObj["realIp"]
                if jsonObj["dataType"] == "clientFile":
                    dataJson = jsonObj["data"]
                    self.processData(dataJson, realIp)
                    
                elif jsonObj["dataType"] == "clientFileData":
                    pass

                elif jsonObj["dataType"] == "clientPicData":
                    pass

                elif jsonObj["dataType"] == "clientPic":
                    pass

                elif jsonObj["dataType"] == "apiData":
                    dataJson = jsonObj["data"]
                    self.processData(dataJson, realIp)

                elif jsonObj["dataType"] == "apiPic":
                    print("msg = " , jsonStr)

                elif jsonObj["dataType"] == "apiPicData":
                    self.collection = self.mongoDB["apiPicData"]
                    jsonItem["realIp"] = realIp
                    for pic in jsonItem["data"][0]["data"]:
                        pic["pic_url"] = None
                        pic["pic_size"] = None
                        self.insertToCollection(jsonItem)
                '''
                realIp = ""
                # tempJsonItem = None
                if not isinstance(jsonList, list):
                    if "dataType" in jsonList.keys() and jsonList["dataType"] == "blobPic":
                        #self.writeToFile("pic", jsonStr)
                        filename = jsonList["filename"]
                        if "data" in jsonList.keys() :
                            #self.writeToBlob(filename, jsonList["data"])
                            pass
                    continue
                #self.writeToFile("data", jsonStr)
                for jsonItem in jsonList:
                    if "realIp" in jsonItem.keys():
                        realIp = jsonItem["realIp"]

                    elif "dataType" in jsonItem.keys() and jsonItem["dataType"] == "encoderData":
                        # Creat collection and insert record to mongoDB
                        self.collection = self.mongoDB["encoderData"]
                        jsonItem["realIp"] = realIp
                        self.insertToCollection(jsonItem)

                    elif "dataType" in jsonItem.keys() and jsonItem["dataType"] == "productionData":
                        self.collection = self.mongoDB["productionData"]
                        jsonItem["realIp"] = realIp
                        self.insertToCollection(jsonItem)

                    elif "dataType" in jsonItem.keys() and jsonItem["dataType"] == "robotData":
                        self.collection = self.mongoDB["robotData"]
                        jsonItem["realIp"] = realIp
                        self.insertToCollection(jsonItem)

                    elif "dataType" in jsonItem.keys() and jsonItem["dataType"] == "clawData":
                        self.collection = self.mongoDB["clawData"]
                        jsonItem["realIp"] = realIp
                        self.insertToCollection(jsonItem)

                    elif "dataType" in jsonItem.keys() and jsonItem["dataType"] == "pcData":
                        self.collection = self.mongoDB["pcData"]
                        jsonItem["realIp"] = realIp
                        self.insertToCollection(jsonItem)

                    elif "dataType" in jsonItem.keys() and jsonItem["dataType"] == "pic":
                        # tempJsonItem = jsonItem
                        
                        self.collection = self.mongoDB["pic"]
                        jsonItem["realIp"] = realIp
                        for pic in jsonItem["data"][0]["data"]:
                            pic["pic_url"] = None
                            pic["pic_size"] = None
                        self.insertToCollection(jsonItem)

                    else :
                        print("Unexpected error! json = " , jsonStr)
'''
            except BaseException as e:
                print("Decoding JSON has failed! json = " , jsonStr, e)
            
        
    
    def getNextId(self):
        result = self.collection.find_one_and_update({"_id": 'info_id'}, {"$inc": {"sequence_value": 1}}, safe=True, new=True)
        if result == None:
            return 0
        nextid = result["sequence_value"]
        return nextid

    def writeToBlob(self, fileName, picBase64Str):
        block_blob_service = BlockBlobService(account_name='ezcloudtest',
         account_key='j8K+ZV0Rc4hn6KaI7ojQELIixnJutUx8fnytL47TPjdRCrAWq3gTRjgrRgFJaUyIdnRUFYSuNfm6bpIZUfa79Q==',
         custom_domain="ezcloudtest.blob.core.chinacloudapi.cn")
        container_name ='test1'
        block_blob_service.create_container(container_name)
        # Set the permission so the blobs are public.
        block_blob_service.set_container_acl(container_name, public_access=PublicAccess.Container)
        base64_data = base64.b64decode(picBase64Str)

        block_blob_service.create_blob_from_bytes(container_name, fileName, base64_data)
        url = block_blob_service.make_blob_url(container_name, fileName)
        blob = block_blob_service._get_blob(container_name, fileName)
        size = blob.properties.content_length
        #print("insert pic to Blob :", url, " blob.properties.content_length = ", blob.properties.content_length)
        #print("insert pic to Blob :", url)
        try:
            self.collection = self.mongoDB["pic"]
            document = self.collection.find_one({"data.data.filename" : fileName})
            
            if document is not None:
                for pic in document["data"][0]["data"]:
                    if pic["filename"] == fileName:
                        pic["pic_url"] = url
                        pic["pic_size"] = size
                updateResult = self.collection.find_one_and_update({"data.data.filename" : fileName},
                               {"$set" : document}, 
                               upsert=True, 
                               return_document=ReturnDocument.AFTER)
                #updateResult = self.collection.update_one({"sequence_value" : document["sequence_value"]}, {"$set" : dict(document)})
                print("updateResult = ", updateResult["sequence_value"], fileName)
        except BaseException as e:
            print("update error!", e, document)
        # modify pic record in mongodb 
        
        #generator = block_blob_service.list_blobs(container_name)
        #for blob in generator:
        #     print("\t Blob properties.content_length: " , blob.properties.content_length)
        #print("insert pic to Blob : " + fileName)
        

def fun_timer():
    operate = KafkaOperate()
    # operate.send()

if __name__ == "__main__":
    # timer = threading.Timer(1, fun_timer)
    # timer.start()
    operate = KafkaOperate()
    operate.receive()
    #operate.send()

